package com.futu.kafka;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.futu.attraction.config.AverageAggregateFunction;
import com.futu.attraction.config.TemperatureAlert;
import com.futu.common.utils.RedisUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.java.Log;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
@RequiredArgsConstructor
@Slf4j
public class TemperatureProcessingService {
    private final StreamExecutionEnvironment env;

    @Resource
    private RedisUtil redisUtil;

    @Resource
    private ApplicationContext applicationContext;

    private static RedisUtil staticRedisUtil;

    // 温度阈值设置为30度，超过此温度会触发告警
    private static final double TEMPERATURE_THRESHOLD = 30.0;

    // 用于存储城市最近的温度数据，用于验证
    private static final ConcurrentHashMap<String, Double> cityTemperatureCache = new ConcurrentHashMap<>();

    // 温度变化阈值，如果温度变化超过此值，可能是异常数据
    private static final double TEMPERATURE_CHANGE_THRESHOLD = 10.0;

    @PostConstruct
    public void init() {
        // 在初始化时将RedisUtil实例赋值给静态变量
        staticRedisUtil = this.redisUtil;
        log.info("初始化RedisUtil静态引用");
        initializeFlinkJob();
    }

    public void initializeFlinkJob(){
        new Thread(()->{
           try {
               log.info("开始初始化Flink作业");
               KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                       .setBootstrapServers("47.113.120.82:9092")
                       .setTopics("temperature")
                       .setGroupId("temperature-flink-group")
                       .setStartingOffsets(OffsetsInitializer.latest())
                       .setValueOnlyDeserializer(new SimpleStringSchema())
                       .build();
               DataStreamSource<String> source = env.fromSource(kafkaSource,
                       WatermarkStrategy.noWatermarks(),
                       "Kafka Source");
               SingleOutputStreamOperator<TemperatureReading> map = source.map(new JsonToTemperatureReadingMapper());

               // 添加温度数据验证过滤器
               SingleOutputStreamOperator<TemperatureReading> validatedData = map.filter(new TemperatureValidationFilter());

               // 过滤高温数据
               SingleOutputStreamOperator<TemperatureReading> filter = validatedData.filter(new HighTemperatureFilter(TEMPERATURE_THRESHOLD));
               SingleOutputStreamOperator<TemperatureAlert> alertStream = filter.map(new TemperatureAlertMapper());

               // 不再将告警信息存入Redis，只打印日志
               alertStream.print("Temperature Alert");

               SingleOutputStreamOperator<Double> streamOperator = filter.map(TemperatureReading::getTemperature)
                       .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).aggregate(new AverageAggregateFunction());
               streamOperator.print("Temperature Alert");
               alertStream.print("Temperature Alert Details");
               streamOperator.print("Average Temperature (5min)");
               try {
                   env.execute("Temperature Processing Job");
               }catch (Exception e)
               {
                   log.error("启动flink作业失败",e);
                   throw new RuntimeException("启动flink作业失败",e);
               }
           }catch (Exception e)
           {
               log.error("处理数据失败",e);
           }
        }).start();
    }

    // 温度数据验证过滤器，过滤掉可能不准确的温度数据
    private static class TemperatureValidationFilter implements FilterFunction<TemperatureReading> {
        @Override
        public boolean filter(TemperatureReading reading) {
            if (reading == null || reading.getCity() == null) {
                log.warn("无效的温度数据: {}", reading);
                return false;
            }

            String city = reading.getCity();
            double temperature = reading.getTemperature();

            // 检查温度是否在合理范围内
            if (temperature < -50 || temperature > 50) {
                log.warn("温度超出合理范围: city={}, temperature={}", city, temperature);
                return false;
            }

            // 检查温度变化是否合理
            Double lastTemperature = cityTemperatureCache.get(city);
            if (lastTemperature != null) {
                double change = Math.abs(temperature - lastTemperature);
                if (change > TEMPERATURE_CHANGE_THRESHOLD) {
                    log.warn("温度变化过大，可能是异常数据: city={}, lastTemp={}, currentTemp={}, change={}",
                            city, lastTemperature, temperature, change);
                    return false;
                }
            }

            // 更新缓存
            cityTemperatureCache.put(city, temperature);
            log.info("温度数据验证通过: city={}, temperature={}", city, temperature);
            return true;
        }
    }

    private static class JsonToTemperatureReadingMapper implements MapFunction<String, TemperatureReading> {
        @Override
        public TemperatureReading map(String json) throws Exception {
            try {
                // 检查是否是简单数字或空字符串
                if (json == null || json.trim().isEmpty()) {
                    log.warn("收到空消息，跳过处理");
                    return createDefaultReading(0);
                }

                // 检查是否是简单数字
                if (json.matches("^\\d+(\\.\\d+)?$")) {
                    log.warn("收到简单数字消息: {}, 创建默认对象", json);
                    double temp = Double.parseDouble(json);
                    return createDefaultReading(temp);
                }

                // 尝试解析JSON
                log.info("收到温度数据: {}", json);
                ObjectMapper mapper = new ObjectMapper();
                mapper.registerModule(new JavaTimeModule());
                TemperatureReading reading = mapper.readValue(json, TemperatureReading.class);
                log.info("解析后的温度数据: city={}, temperature={}", reading.getCity(), reading.getTemperature());
                return reading;
            } catch (Exception e) {
                // 记录错误但不中断处理
                log.error("消息解析失败: {}, 错误: {}", json, e.getMessage());
                // 返回默认对象，避免流处理中断
                return createDefaultReading(0);
            }
        }

        private TemperatureReading createDefaultReading(double temp) {
            TemperatureReading reading = new TemperatureReading();
            reading.setCity("Unknown");
            reading.setTemperature(temp);
            reading.setTimestamp(new Date());
            return reading;
        }
    }

    private static class HighTemperatureFilter implements FilterFunction<TemperatureReading> {
        private final double threshold;

        public HighTemperatureFilter(double threshold) {
            this.threshold = threshold;
        }

        @Override
        public boolean filter(TemperatureReading reading) {
            boolean isHighTemperature = reading.getTemperature() > threshold;
            if (isHighTemperature) {
                log.info("检测到高温: city={}, temperature={}", reading.getCity(), reading.getTemperature());
            }
            return isHighTemperature;
        }
    }

    private static class TemperatureAlertMapper implements MapFunction<TemperatureReading, TemperatureAlert> {
        @Override
        public TemperatureAlert map(TemperatureReading reading) {
            String message;
            double temp = reading.getTemperature();

            if (temp > 35) {
                message = "温度过高，注意防暑降温";
            } else if (temp > 30) {
                message = "温度较高，请适当补充水分";
            } else if (temp > 25) {
                message = "温度舒适，适合户外活动";
            } else if (temp > 10) {
                message = "温度适宜，建议适当增减衣物";
            } else if (temp > 0) {
                message = "温度较低，注意保暖";
            } else {
                message = "温度很低，请注意防寒保暖";
            }

            TemperatureAlert alert = new TemperatureAlert(
                    reading.getCity(),
                    reading.getTemperature(),
                    new Date(),
                    message
            );

            // 设置告警状态
            alert.setAlert(true);

            log.info("生成温度告警: {}", JSON.toJSONString(alert));
            return alert;
        }
    }
}
